package com.xxx.rpc.registry.curator;

import java.util.concurrent.CountDownLatch;

import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.ZKPaths;

import com.xxx.rpc.registry.zookeeper.Constant;

public class PublishTest {
	private CuratorFramework client = null;
	private CountDownLatch countDownLatch = new CountDownLatch(1);

	private String zkAddress = "127.0.0.1:2181";

	/**
	 * 客户端初始化
	 * 
	 * @throws Exception
	 */
	public CuratorFramework createClient() throws Exception {
		client = CuratorFrameworkFactory.builder().connectString(zkAddress)
				.retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))
				.connectionTimeoutMs(Constant.ZK_CONNECTION_TIMEOUT).sessionTimeoutMs(Constant.ZK_SESSION_TIMEOUT)
				.build();
		// 客户端注册监听，进行连接配置
		client.getConnectionStateListenable().addListener(connectionListener);
		client.start();
		// 连接成功后，才进行下一步的操作
		countDownLatch.await();
		
		return this.client;
	}

	public void closeClient(){
		if(this.client != null && this.client.getState() != CuratorFrameworkState.STOPPED){
			this.client.close();
			this.client = null;
		}
	}
	
	/**
	 * 客户端重新注册
	 */
	private void reinit() {
		System.err.println("need reinit....");
		try {
			this.client = createClient();
			
			//watchDirectPath(directPath);
			//watchTreePath(treePath);
			//FIXME 罗珊  观察下  不注册 能继续观察到不？？？
			System.out.println("FIXME 罗珊  观察下  不注册 能继续观察到不？？？");
		} catch (Exception e) {
			System.err.println("failed reinit,"+ e.getMessage());
			this.client =  null;
		}
	}

	 

	public static void main(String[] args) throws Exception {
		PublishTest pp = new PublishTest();
		pp.createClient();
		System.out.println("init ok........");
		
		pp.watchDirectPath(Constant.ZK_REGISTRY_PATH);
		
		pp.watchTreePath(Constant.ZK_REGISTRY_PATH);
		
		System.out.println("pp.watch ("+Constant.ZK_REGISTRY_PATH+") ... ");
		Thread.sleep(Integer.MAX_VALUE);
	}

	private String directPath ;
	private String treePath ;
	private PathChildrenCache pathCache ;
	private TreeCache treeCache;
	
	// 对path进行监听配置
	public void watchDirectPath(String directPath) throws Exception {
		this.directPath = directPath;
		// 子节点的监听
		this.pathCache = new PathChildrenCache(client, directPath, false);
		pathCache.start();
		// 注册监听
		pathCache.getListenable().addListener(plis);
		

	}
	
	// 对path进行监听配置
	public void watchTreePath(String treePath) throws Exception {
			this.treePath = treePath;
			 
			// 对path路径下所有孩子节点的监听
			this.treeCache = new TreeCache(client, treePath);
			treeCache.start();
			treeCache.getListenable().addListener(treeCacheListener);
		}

	// 客户端的监听配置
	private ConnectionStateListener connectionListener = new ConnectionStateListener() {
		public void stateChanged(CuratorFramework client, ConnectionState newState) {
			if (newState == ConnectionState.CONNECTED) {
				System.out.println("connected established");
				countDownLatch.countDown();
			} else if (newState == ConnectionState.LOST) {
				
				System.out.println("connection lost,waiting for reconection");
				try {
					PublishTest.this.closeClient();
					
					while(PublishTest.this.client == null){
						System.out.println("reinit---");
						reinit();
						Thread.sleep(1000);
						System.out.println("inited---");
					}
					
				} catch (Exception e) {
					System.err.println("re-inited failed");
				}
			}

		}
	};
	
	 

	/**
	 * 子节点的监听
	 */
	private PathChildrenCacheListener plis = new PathChildrenCacheListener() {

		public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
			switch (event.getType()) {
			case CHILD_ADDED: {
				System.out.println("PathChildrenCacheListener Node added: " + ZKPaths.getNodeFromPath(event.getData().getPath()) + " , data:"
						+ ss.deserialize(event.getData().getData()) );
				break;
			}

			case CHILD_UPDATED: {
				System.out.println("PathChildrenCacheListener Node changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()) + " , data: "
						+ ss.deserialize(event.getData().getData()) );
				break;
			}

			case CHILD_REMOVED: {
				System.out.println("PathChildrenCacheListener Node removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()) );
				break;
			}
			default:
				System.out.println("PathChildrenCacheListener default...." + ZKPaths.getNodeFromPath(event.getData().getPath()) );
				break;
			}

		}
	};
	
	
	private SerializableSerializer ss = new SerializableSerializer();
	/**
	 * 所有子节点的监听
	 */
	private TreeCacheListener treeCacheListener = new TreeCacheListener() {

		public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
			switch (event.getType()) {
			case NODE_ADDED:
				if(Constant.ZK_REGISTRY_PATH.equals(event.getData().getPath())){
					System.err.println("ignore path ：" +event.getData().getPath());
					break;
				}
				System.out.println("TreeCacheListener TreeNode added: " + event.getData().getPath() + " , data: "  + ss.deserialize(event.getData().getData()));
				break;
			case NODE_UPDATED:
				System.out.println("TreeCacheListener TreeNode updated: " + event.getData().getPath() + " , data: " + ss.deserialize(event.getData().getData()));
				break;
			case NODE_REMOVED:
				System.out.println("TreeCacheListener TreeNode removed: " + event.getData().getPath());
				break;
			default:
				System.out.println("TreeCacheListener TreeNode : " + event.getType());
				break;
			}
		}
	};
}
